package cn.sciento.transfer.app.service.impl;

import com.alibaba.fastjson.JSON;
import cn.sciento.core.domain.Page;
import io.choerodon.mybatis.pagehelper.domain.PageRequest;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.stream.Collectors;
import org.apache.commons.lang3.exception.ExceptionUtils;
import cn.sciento.transfer.api.dto.DataCheckResultDTO;
import cn.sciento.transfer.api.dto.DataChkBatchCreateDTO;
import cn.sciento.transfer.api.dto.DiffRowDTO;
import cn.sciento.transfer.api.dto.UpdateRowRecordDTO;
import cn.sciento.transfer.app.service.DataChkBatchService;
import cn.sciento.transfer.domain.entity.DataChkBatch;
import cn.sciento.transfer.domain.entity.DataChkBatchDtl;
import cn.sciento.transfer.domain.entity.DataChkBatchLine;
import cn.sciento.transfer.domain.repository.DataChkBatchDtlRepository;
import cn.sciento.transfer.domain.repository.DataChkBatchLineRepository;
import cn.sciento.transfer.domain.repository.DataChkBatchRepository;
import cn.sciento.transfer.infra.datasource.sql.FieldBuilder;
import cn.sciento.transfer.infra.datasource.sql.SqlUtils;
import cn.sciento.transfer.infra.mapper.DataChkBatchMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.core.task.AsyncTaskExecutor;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;

@Service
public class DataChkBatchServiceImpl implements DataChkBatchService {
  private static final Logger LOGGER = LoggerFactory.getLogger(DataChkBatchServiceImpl.class);
  
  @Autowired
  private DataChkBatchLineRepository dataChkBatchLineRepository;
  
  @Autowired
  private DataChkBatchRepository dataChkBatchRepository;
  
  @Autowired
  private DataChkBatchDtlRepository dataChkBatchDtlRepository;
  
  @Autowired
  @Qualifier("transferDataCheckExecutor")
  private AsyncTaskExecutor asyncTaskExecutor;
  
  @Autowired
  private DataCheckService dataCheckService;
  
  @Autowired
  private DataChkBatchMapper dataChkBatchMapper;
  
  public DataChkBatchLine detailDataChk(Long dataChkBatchLineId) {
    DataChkBatchLine dataChkBatchLine = new DataChkBatchLine();
    dataChkBatchLine.setDataChkBatchLineId(dataChkBatchLineId);
    List<DataChkBatchLine> dataChkBatchLineList = this.dataChkBatchMapper.listDataChk(dataChkBatchLine);
    Assert.isTrue((!CollectionUtils.isEmpty(dataChkBatchLineList) && dataChkBatchLineList.size() == 1), "error.data_invalid");
    return dataChkBatchLineList.get(0);
  }
  
  @Transactional(rollbackFor = {Exception.class})
  public DataChkBatch createChk(DataChkBatchCreateDTO dataChkBatchCreateDTO) {
    List<DataChkBatchLine> allNeedCheckConfig = new ArrayList<>();
    if ("ALL".equals(dataChkBatchCreateDTO.getChkLevel())) {
      DataChkBatchLine queryAllConfig = new DataChkBatchLine();
      queryAllConfig.setChkLevel(dataChkBatchCreateDTO.getChkLevel());
      allNeedCheckConfig.addAll(this.dataChkBatchLineRepository.plainAllConfig(queryAllConfig));
    } else if (!CollectionUtils.isEmpty(dataChkBatchCreateDTO.getDataChkBatchLineList())) {
      dataChkBatchCreateDTO.getDataChkBatchLineList().forEach(item -> {
            item.setChkLevel(dataChkBatchCreateDTO.getChkLevel());
            item.validParam();
            allNeedCheckConfig.addAll(this.dataChkBatchLineRepository.plainAllConfig(item));
          });
    } 
    dataChkBatchCreateDTO.setDataChkBatchLineList(allNeedCheckConfig);
    DataChkBatch dataChkBatch = new DataChkBatch();
    dataChkBatch.setChkLevel(dataChkBatchCreateDTO.getChkLevel());
    Date date = new Date();
    SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyyMMddHHmmss");
    dataChkBatch.setBatchNum(simpleDateFormat.format(date));
    this.dataChkBatchRepository.insertSelective(dataChkBatch);
    List<DataChkBatchLine> dataChkBatchLines = dataChkBatchCreateDTO.getDataChkBatchLineList();
    if (!CollectionUtils.isEmpty(dataChkBatchLines)) {
      dataChkBatchLines.forEach(item -> item.setDataChkBatchId(dataChkBatch.getDataChkBatchId()));
      this.dataChkBatchLineRepository.batchInsertSelective(dataChkBatchLines);
    } 
    Iterator<DataChkBatchLine> iterator = dataChkBatchLines.iterator();
    while (iterator.hasNext()) {
      DataChkBatchLine item = iterator.next();
      this.asyncTaskExecutor.execute(() -> {
            try {
              checkAndSaveDiff(item);
            } catch (Exception e) {
              item.setProcessTime(new Date());
              item.setProcessMsg(ExceptionUtils.getStackTrace(e));
              item.setProcessStatus("E");
              this.dataChkBatchLineRepository.updateOptional(item, new String[] { "processMsg", "processStatus", "processTime" });
              if (LOGGER.isDebugEnabled()) {
                  LOGGER.debug("check data error,", e);
              }
            } 
          });
      iterator.remove();
    } 
    return dataChkBatch;
  }
  
  public Page<DataChkBatchDtl> pageDataCheckDtl(Long dataChkBatchId, Long dataChkBatchLineId, PageRequest pageRequest) {
    DataChkBatchDtl dataChkBatchDtl = new DataChkBatchDtl();
    dataChkBatchDtl.setDataChkBatchId(dataChkBatchId);
    dataChkBatchDtl.setDataChkLineId(dataChkBatchLineId);
    Page<DataChkBatchDtl> page = this.dataChkBatchDtlRepository.pageAndSort(pageRequest, dataChkBatchDtl);
    if (!CollectionUtils.isEmpty(page.getContent())) {
      DataChkBatch dataChkBatch = (DataChkBatch)this.dataChkBatchRepository.selectByPrimaryKey(dataChkBatchId);
      Assert.notNull(dataChkBatch, "error.data_invalid");
      page.getContent().forEach(item -> item.setBatchNum(dataChkBatch.getBatchNum()));
    } 
    return page;
  }
  
  private void checkAndSaveDiff(DataChkBatchLine item) {
    item.setProcessTime(new Date());
    item.setProcessStatus("R");
    this.dataChkBatchLineRepository.updateOptional(item, new String[] { "processMsg", "processStatus", "processTime" });
    DataCheckResultDTO dataCheckResultDTO = this.dataCheckService.check(item.getSourceDs(), item
        .getSourceService(), item
        .getSourceDb(), item
        .getSourceTable(), item
        .getTargetDs(), item
        .getTargetService(), item
        .getTargetDb(), item
        .getTargetTable(), item
        .getTenantId());
    String masterKeyFiled = FieldBuilder.formatFieldToColumn(SqlUtils.getTablePrimaryKey(item.getSourceTable(), item
          .getSourceDs(), item
          .getSourceDb(), item
          .getSourceService()).get(0));
    List<Map<String, Object>> insertList = dataCheckResultDTO.getMasterInsertList();
    if (!CollectionUtils.isEmpty(insertList)) {
      List<DataChkBatchDtl> chkBatchDtlInsertList = (List<DataChkBatchDtl>)insertList.stream().map(row -> {
            DataChkBatchDtl dataChkBatchDtl = new DataChkBatchDtl();
            dataChkBatchDtl.setDataChkBatchId(item.getDataChkBatchId());
            dataChkBatchDtl.setDataChkLineId(item.getDataChkBatchLineId());
            dataChkBatchDtl.setDataId(Long.valueOf(Long.parseLong(row.get(masterKeyFiled).toString())));
            dataChkBatchDtl.setDiffType("NO_DATA_LINE");
            dataChkBatchDtl.setSourceLine(JSON.toJSONString(row));
            return dataChkBatchDtl;
          }).collect(Collectors.toList());
      this.dataChkBatchDtlRepository.batchInsertSelective(chkBatchDtlInsertList);
    } 
    if (!CollectionUtils.isEmpty(dataCheckResultDTO.getMasterDeleteList())) {
      List<DataChkBatchDtl> chkBatchDtlDeleteList = (List<DataChkBatchDtl>)dataCheckResultDTO.getMasterDeleteList().stream().map(row -> {
            DataChkBatchDtl dataChkBatchDtl = new DataChkBatchDtl();
            dataChkBatchDtl.setDataChkBatchId(item.getDataChkBatchId());
            dataChkBatchDtl.setDataChkLineId(item.getDataChkBatchLineId());
            dataChkBatchDtl.setDataId(Long.valueOf(Long.parseLong(row.get(masterKeyFiled).toString())));
            dataChkBatchDtl.setDiffType("DELETED_DATA_LINE");
            dataChkBatchDtl.setSourceLine("");
            dataChkBatchDtl.setTargetLine(JSON.toJSONString(row));
            return dataChkBatchDtl;
          }).collect(Collectors.toList());
      this.dataChkBatchDtlRepository.batchInsertSelective(chkBatchDtlDeleteList);
    } 
    if (!CollectionUtils.isEmpty(dataCheckResultDTO.getUpdateRows())) {
      List<DataChkBatchDtl> chkBatchDtlUpdateList = (List<DataChkBatchDtl>)dataCheckResultDTO.getUpdateRows().stream().map(row -> {
            DataChkBatchDtl dataChkBatchDtl = new DataChkBatchDtl();
            dataChkBatchDtl.setDataChkBatchId(item.getDataChkBatchId());
            dataChkBatchDtl.setDataChkLineId(item.getDataChkBatchLineId());
            dataChkBatchDtl.setDataId(Long.valueOf(Long.parseLong(row.getMasterUpdateRow().get(masterKeyFiled).toString())));
            dataChkBatchDtl.setDiffType("FIELD_VALUE_DIFF");
            List<DiffRowDTO> diffRowDTOList = new ArrayList<>();
            Map<String, Object> masterRow = row.getMasterUpdateRow();
            Map<String, Object> slaveRow = row.getSlaveUpdateRow();
            for (Map.Entry<String, Object> entry : masterRow.entrySet()) {
              if (!Objects.equals(entry.getValue(), slaveRow.get(entry.getKey()))) {
                DiffRowDTO diffRowDTO = new DiffRowDTO();
                diffRowDTO.setFieldName(entry.getKey());
                diffRowDTO.setSourceValue(entry.getValue());
                diffRowDTO.setTargetValue((slaveRow.get(entry.getKey()) == null) ? "null" : slaveRow.get(entry.getKey()));
                diffRowDTOList.add(diffRowDTO);
              } 
            } 
            dataChkBatchDtl.setDiffContent(JSON.toJSONString(diffRowDTOList));
            dataChkBatchDtl.setSourceLine(JSON.toJSONString(row.getMasterUpdateRow()));
            dataChkBatchDtl.setTargetLine(JSON.toJSONString(row.getSlaveUpdateRow()));
            return dataChkBatchDtl;
          }).collect(Collectors.toList());
      this.dataChkBatchDtlRepository.batchInsertSelective(chkBatchDtlUpdateList);
    } 
    item.setProcessTime(new Date());
    item.setProcessStatus("S");
    item.setProcessMsg(JSON.toJSONString(dataCheckResultDTO.getDataCheckHandleMsgDTO()));
    this.dataChkBatchLineRepository.updateOptional(item, new String[] { "processMsg", "processStatus", "processTime" });
  }
}
